package com.gugee.ins.data.blogger.statistics.listener;

import com.gugee.ins.data.blogger.statistics.config.BloggerStatisticsAppConfig;
import com.gugee.ins.data.blogger.statistics.core.*;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.common.msg.InsMBIdenticalLevelMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class StBloggerListener {


    protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    StBloggerRealFollowerService bloggerRealFollowerService;

    @Autowired
    StBloggerActiveFollowerDataService stBloggerActiveFollowerDataService;

    @Autowired
    StIdenticalLevelBloggerService stIdenticalLevelBloggerService;

    @Autowired
    StBloggerMediaAdService stBloggerMediaAdService;

    @Autowired
    StProductBaseService stProductBaseService;


    @Autowired
    BloggerStatisticsAppConfig bloggerStatisticsAppConfig;

    private BloggerRealFollowerSt bloggerRealFollowerSt = new BloggerRealFollowerSt();
    private BloggerActiveFollowerSt bloggerActiveFollowerSt = new BloggerActiveFollowerSt();
    private BloggerMediaAdSt bloggerMediaAdSt = new BloggerMediaAdSt();
    private BloggerMainIdenticalLevelSt bloggerMainIdenticalLevelSt = new BloggerMainIdenticalLevelSt();

    private ProductBaseSt productBaseSt = new ProductBaseSt();

    /**
     * 真实粉丝统计
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBloggerRealFollowerSt.id}",
            topics = {"${kafka.topics.insMBloggerRealFollowerSt.online}"},
            containerFactory = "factoryInsBloggerStAck")
    public void bloggerRealFollowerStListen(List<InsBloggerMessage> messages, Acknowledgment ack) {
        bloggerRealFollowerSt.handle("BloggerRealFollowerSt", messages,bloggerStatisticsAppConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    /**
     * 活跃粉丝数据统计
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBloggerActiveFollowerDataSt.id}",
            topics = {"${kafka.topics.insMBloggerActiveFollowerDataSt.online}","${kafka.topics.insMBloggerActiveFollowerDataSt.priority}"},
            containerFactory = "factoryInsBloggerStAck")
    public void bloggerActiveFollowerStListen(List<InsBloggerMessage> messages, Acknowledgment ack) {
        bloggerActiveFollowerSt.handle("BloggerActiveFollowerDataSt", messages,bloggerStatisticsAppConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    /**
     * 博主广告贴子统计
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBloggerMediaAdSt.id}",
            topics = {"${kafka.topics.insMBloggerMediaAdSt.online}","${kafka.topics.insMBloggerMediaAdSt.priority}"},
            containerFactory = "factoryInsBloggerStAck")
    public void bloggerMediaAdStListen(List<InsBloggerMessage> messages,Acknowledgment ack){
        bloggerMediaAdSt.handle("BloggerMediaAdSt",messages,bloggerStatisticsAppConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    /**
     * 统计同分类同量级粉丝博主平均值
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBloggerIdenticalLevelSt.id}",topics = {"${kafka.topics.insMBloggerIdenticalLevelSt.online}"},containerFactory = "factoryInsMBIdenticalLevelAck")
    public void bloggerMainIdenticalLevelStListen(List<InsMBIdenticalLevelMessage> messages,Acknowledgment ack){
        bloggerMainIdenticalLevelSt.handle("BloggerMediaIdenticalLevelSt",messages,bloggerStatisticsAppConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    /**
     * 商品统计
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insProductBaseSt.id}",
            topics = {"${kafka.topics.insProductBaseSt.online}"},
            containerFactory = "factoryInsBloggerStAck")
    public void productBaseStListen(List<InsBloggerMessage> messages,Acknowledgment ack){
        productBaseSt.handle("insProductBaseSt",messages,bloggerStatisticsAppConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    class BloggerRealFollowerSt extends AbstractGugeeConsumerHandler<InsBloggerMessage>{

        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            bloggerRealFollowerService.handler(messages);
        }
    }

    class BloggerActiveFollowerSt extends AbstractGugeeConsumerHandler<InsBloggerMessage>{

        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            stBloggerActiveFollowerDataService.handler(messages);
        }
    }

    class BloggerMediaAdSt extends AbstractGugeeConsumerHandler<InsBloggerMessage>{

        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            stBloggerMediaAdService.handler(messages);
        }
    }

    class BloggerMainIdenticalLevelSt extends AbstractGugeeConsumerHandler<InsMBIdenticalLevelMessage>{

        @Override
        public void dealMsg(List<InsMBIdenticalLevelMessage> messages) {
            stIdenticalLevelBloggerService.handler(messages);
        }
    }

    class ProductBaseSt extends AbstractGugeeConsumerHandler<InsBloggerMessage>{

        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            stProductBaseService.handler(messages);
        }
    }
}
